package avicit.bdp.dcs.semistructure.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.dto.TaskInstanceDto;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.HiveUtils;
import avicit.bdp.common.utils.database.IotDbSessionUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.job.service.DcsJobService;
import avicit.bdp.dcs.semistructure.dao.JobFileDao;
import avicit.bdp.dcs.semistructure.dto.HiveJsonDto;
import avicit.bdp.dcs.semistructure.dto.JobFileDTO;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.task.service.TaskService;
import avicit.bdp.dcs.utils.DatabaseUtil;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dcs.utils.JdbcConstants;
import com.alibaba.fastjson2.JSON;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * @author liyb 任务文件实现类
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class JobFileService extends BaseService<JobFileDao, JobFileDTO> {

  private static final Logger logger = LoggerFactory.getLogger(JobFileService.class);

  @Autowired private DcsJobService jobService;
  @Autowired private DcsFileService fileService;
  @Autowired private TaskService taskService;
  @Autowired private DataSourceUtils dataSourceUtils;

  /**
   * 删除iotdb中文件对应的数据
   *
   * @param filePath filePath
   * @return 删除结果
   */
  public Boolean deleteFileData(String filePath) throws Exception {
    JobFileDTO jobFileQuery = new JobFileDTO();
    String defaultFS = ConfigUtils.getInstance().getString(Constants.FS_DEFAULTFS);
    jobFileQuery.setFilePath(defaultFS + filePath);
    List<JobFileDTO> jobFileList = this.selectList(jobFileQuery);
    if (CollectionUtils.isEmpty(jobFileList)) {
      logger.info("没有作业中引用了该文件：{}", filePath);
      return true;
    }

    for (JobFileDTO jobFileDTO : jobFileList) {
      JobDTO jobDTO = jobService.selectByPrimaryKey(jobFileDTO.getJobId());
      CheckParaUtils.checkDbObjectValidity(
          jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobFileDTO.getJobId()));

      TaskDTO taskDTO = taskService.selectByPrimaryKey(jobFileDTO.getTaskId());
      CheckParaUtils.checkDbObjectValidity(
          taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), jobFileDTO.getTaskId()));
      TaskInstanceDto taskInstanceDto =
          jobService.getTaskInstanceByProcessInstanceId(
              jobFileDTO.getProcessInstanceId(), jobFileDTO.getTaskId());

      if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
        JobDatasource dstDatasource =
            dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
        if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
          String deviceId = taskDTO.getDstTableName();

          // 删除iotdb数据: delete from root.liyb.test where time>=1 and time<=5
          String sqlDeleteData =
              "delete from "
                  + deviceId
                  + "  where time>="
                  + taskInstanceDto.getStartTime()
                  + " and time<="
                  + taskInstanceDto.getEndTime();
          logger.info("删除iotdb数据sql:{}", sqlDeleteData);
          IotDbSessionUtils.initIotdbSessionPool(
              dstDatasource.getJdbcUrl(),
              dstDatasource.getJdbcUsername(),
              dstDatasource.getEncryptJdbcPassword());

          Thread.sleep(30 * 1000);
          IotDbSessionUtils.deleteIotdbData(sqlDeleteData);
        } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
          // 清空hive表数据
          HiveJsonDto hiveJsonDto = JSON.parseObject(taskDTO.getHiveJson(), HiveJsonDto.class);
          String hiveTableLocation =
              HiveUtils.getHiveTableLocation(
                  DatabaseUtil.getConnection(dstDatasource), hiveJsonDto.getDstTable());
          String hiveFilePath = hiveTableLocation + "/" + hiveJsonDto.getFileName();
          fileService.deleteFileByPath(hiveFilePath);

          //                    TaskDTO taskQuery = new TaskDTO();
          //                    taskQuery.setJobId(jobDTO.getId());
          //                    List<TaskDTO> taskList = taskService.selectList(taskQuery);
          //                    if (CollectionUtils.isNotEmpty(taskList)) {
          //                        for (TaskDTO task : taskList) {
          //                            HiveJsonDto hiveJsonDto =
          // JSONArray.parseObject(task.getHiveJson(), HiveJsonDto.class);
          //                            String hiveTableLocation =
          // HiveUtils.getHiveTableLocation(DatabaseUtil.getConnection(dstDatasource),
          // hiveJsonDto.getDstTable());
          //                            String hiveFilePath = hiveTableLocation + "/" +
          // hiveJsonDto.getFileName();
          //                            fileService.deleteFileByPath(hiveFilePath);
          //                        }
          //                    }
        }
      }

      // 记录日志
      BdpLogUtil.log4Delete(jobFileDTO);

      // 删除bdp_dcs_job_file
      this.deleteByPrimaryKey(jobFileDTO.getId());
      logger.info("delete file data success: {}", jobFileDTO.getFilePath());
    }
    return true;
  }

  /**
   * 统计设备数据总数
   *
   * @param filePath
   * @return
   * @throws Exception
   */
  public Long countDeviceData(String filePath) throws Exception {
    JobFileDTO jobFileQuery = new JobFileDTO();
    String defaultFS = ConfigUtils.getInstance().getString(Constants.FS_DEFAULTFS);
    jobFileQuery.setFilePath(defaultFS + filePath);
    List<JobFileDTO> jobFileList = this.selectList(jobFileQuery);
    if (CollectionUtils.isEmpty(jobFileList)) {
      logger.info("没有作业中引用了该文件：{}", filePath);
      return 0L;
    }

    long totalRecords = 0L;
    for (JobFileDTO jobFileDTO : jobFileList) {
      JobDTO jobDTO = jobService.selectByPrimaryKey(jobFileDTO.getJobId());
      CheckParaUtils.checkDbObjectValidity(
          jobDTO, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobFileDTO.getJobId()));

      TaskDTO taskDTO = taskService.selectByPrimaryKey(jobFileDTO.getTaskId());
      CheckParaUtils.checkDbObjectValidity(
          taskDTO, logger, String.format(DcsErrorCode.TASK_NULL.getDesc(), jobFileDTO.getTaskId()));
      TaskInstanceDto taskInstanceDto =
          jobService.getTaskInstanceByProcessInstanceId(
              jobFileDTO.getProcessInstanceId(), jobFileDTO.getTaskId());

      if (jobDTO.getJobType() == JobType.SEMISTRUCTURE.getCode()) {
        JobDatasource dstDatasource =
            dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
        if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
          String deviceId = taskDTO.getDstTableName();

          // 统计iotdb数据: select count(*) from root.liyb.test where time>=1615876833000 and
          // time<=1615876843000
          String sqlCountData =
              "select count(*) from "
                  + deviceId
                  + " where time>="
                  + taskInstanceDto.getStartTime()
                  + " and time<="
                  + taskInstanceDto.getEndTime();
          logger.info("统计iotdb数据sql:{}", sqlCountData);
          IotDbSessionUtils.initIotdbSessionPool(
              dstDatasource.getJdbcUrl(),
              dstDatasource.getJdbcUsername(),
              dstDatasource.getEncryptJdbcPassword());
          totalRecords += IotDbSessionUtils.countDeviceData(sqlCountData);
        }
      }
    }
    return totalRecords;
  }
}
